package com.shujia.flink.tf

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala._

object Demo1Map {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val studentDS: DataStream[String] = env.readTextFile("data/students.txt")


    /**
      * map 算子
      * 1、传入一个函数 -- scala api
      * 2、传入一个MapFunction  -- java api
      *
      */
    val mapDS: DataStream[(String, Int)] = studentDS.map(new MapFunction[String, (String, Int)] {

      /**
        *
        * @param value : ds中的一行数据
        * @return 返回数据的类型
        */
      override def map(value: String): (String, Int) = {
        val clazz: String = value.split(",")(4)
        (clazz, 1)
      }

    })

    mapDS
      .keyBy(_._1)
      .sum(1)
      .print()


    env.execute()

  }

}

